package com.tensai.study.新概念.响应式编程.reactor.flux;

import org.junit.Test;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.HashSet;

public class BufferTest {

    @Test
    public void test() {
        //每次缓存一定数量的元素到List buckets里，并push出去
        Flux.range(1, 30).buffer(20).subscribe(System.out::println);
        System.out.println("---------- 分割线1 ----------");

        //每次缓存一定数量，并跳过一定数量的元素到List buckets里，并push出去
        Flux.range(1, 30).buffer(10, 10).subscribe(System.out::println);
        System.out.println("---------- 分割线2 ----------");

        //每次缓存一定数量，并跳过一定数量的元素到指定的Set buckets里，并push出去
        Flux.range(1, 30).buffer(20, 20, HashSet::new).subscribe(System.out::println);
        System.out.println("---------- 分割线3 ----------");

        //指定时间内，每次缓存一定数量的元素到List buckets里，并push出去
        Flux.interval(Duration.ofMillis(100)).buffer(500).take(3).toStream().forEach(System.out::println);
        System.out.println("---------- 分割线4 ----------");

        //缓存元素到List buckets里当符合条件时，把元素push出去
        Flux.range(1, 10).bufferUntil(i -> i % 2 == 0).subscribe(System.out::println);
        System.out.println("---------- 分割线5 ----------");

        //把符合条件的缓存元素到List buckets里并把元素push出去
        Flux.range(1, 10).bufferWhile(i -> i % 2 == 0).subscribe(System.out::println);


    }
}
